package com.ww.flink

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

//找到发生加速的车辆
//同一车牌号 下一次速度减去当次速度大于30则视为急加速
//数据输入格式:车牌号 speed
object Flink_try09_value_state {
  case class CarInfo(carId:String,speed:Long)
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val initStream = env.socketTextStream("node01",8888)
    val stream = initStream.map(line=>{
      val arr: Array[String] = line.split(" ")
      CarInfo(arr(0),arr(1).toLong)
    }).keyBy(_.carId).map(new RichMapFunction[CarInfo,String] {
      private var lastSpeed:ValueState[Long] = _ //记录上次的数据
      override def open(parameters: Configuration) = {
        val desc = new ValueStateDescriptor[Long]("lastSpeed", createTypeInformation[Long])
        lastSpeed = getRuntimeContext.getState(desc)
      }

      override def map(in: CarInfo) = {

        val lasted: Long = lastSpeed.value()
        println(lastSpeed.value())
        println(lasted)
        lastSpeed.update(in.speed)
        if(lasted!=0&&in.speed-lasted>30){
          s"超速:${in.toString},之前速度为:$lasted,速度只差为:${in.speed-lasted}"
        }else{
          s"未超速:${in.toString},之前速度为:$lasted,速度只差为:${in.speed-lasted}"
        }

      }
    }).print()

    env.execute()
  }
}
